package com.leilei;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/13 21:07
 * @desc flink jdbc 连接器
 */
public class FlinkConnectorJDBC {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        DataStreamSource<VehicleAlarm> source = env.addSource(new MySource());
        String sql = "insert into vehicle_alarm_202103 (`id`,`license_plate`,`plate_color`,`device_time`,`zone`) " +
                "values(?,?,?,?,?)";
        JdbcConnectionOptions jdbcBuild = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName("com.mysql.jdbc.Driver")
                .withUrl("jdbc:mysql://xxxx:3306/alarm-sc?useUnicode=true&characterEncoding=utf-8&useSSL=false")
                .withUsername("root")
                .withPassword("root")
                .build();
        /**
         * 使用JDBC连接器 将数据发送到Mysql
         */
        source.addSink(JdbcSink.sink(sql, (ps, vehicle) -> {
            ps.setString(1, vehicle.getId());
            ps.setString(2, vehicle.getLicensePlate());
            ps.setString(3, vehicle.getPlateColor());
            ps.setLong(4, vehicle.getDeviceTime());
            ps.setString(5, vehicle.getZone());
        }, jdbcBuild));
        try {
            env.execute("jdbc-connector");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class VehicleAlarm {
        private String id;
        private String licensePlate;
        private String plateColor;
        private Long deviceTime;
        private String zone;
    }

    /**
     * 自定义数据源
     */
    public static class MySource extends RichSourceFunction<VehicleAlarm> {
        /**
         * 运行时 调用逻辑
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<VehicleAlarm> ctx) throws Exception {
                long id = System.currentTimeMillis()/1000;
                VehicleAlarm vehicleAlarm = new VehicleAlarm(String.valueOf(id), "川A" + id,
                        "紫", System.currentTimeMillis(), "sc");
                ctx.collect(vehicleAlarm);
                Thread.sleep(10000);
            }

        /**
         * 结束时调用逻辑
         */
        @Override
        public void cancel() {

        }
    }

}
